package cn.itcast.flink.async;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.util.Collections;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

/**
 * 定义子类，实现函数接口：AsyncFunction，重写asyncInvoke方法，实现异步请求数据库，获取数据
 *
 * @author lilulu
 * @date 2023-04-08 23:21
 */
public class AsyncMysqlRequest extends RichAsyncFunction<Tuple2<String, String>, String> {

    private Connection connection;
    private PreparedStatement preparedStatement = null;
    private ResultSet resultSet = null;
    //定义线程池变量
    private ExecutorService executorService = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        //初始化线程池
        executorService = Executors.newFixedThreadPool(10);
        Class.forName("com.mysql.jdbc.Driver");
        connection = DriverManager.getConnection("jdbc:mysql://node1:3306/?useSSL=false", "root", "123456");
        preparedStatement = connection.prepareStatement("SELECT user_name FROM db_flink.tbl_user_info WHERE user_id = ?");
        super.open(parameters);
    }

    @Override
    public void asyncInvoke(Tuple2<String, String> inputTuple2, ResultFuture<String> resultFuture) throws Exception {
        /*
            input -> 数据流中每条数据： (u_1009,     u_1009,click,2022-08-06 19:30:55.347)
                        |
                   wujiu,u_1009,click,2022-08-06 19:30:55.347
         */
        String userId = inputTuple2.f0;
//        通过线程池请求Mysql数据库，到达异步请求效果
        Future<String> future = executorService.submit(
                new Callable<String>() {
                    @Override
                    public String call() throws Exception {
                        preparedStatement.setString(1, userId);
                        resultSet = preparedStatement.executeQuery();
                        String userName = "未知";
                        while (resultSet.next()) {
                            userName = resultSet.getString("user_name");
                        }
                        return userName;
                    }
                }
        );

        // 获取异步请求结果,将查询数据结构异步返回
        String userName = future.get();
        String outPut = userName + "," + inputTuple2.f1;
        resultFuture.complete(Collections.singletonList(outPut));
    }


    @Override
    public void timeout(Tuple2<String, String> input, ResultFuture<String> resultFuture) throws Exception {
        String log = input.f1;
        String outPut = "unknown" + log;
        resultFuture.complete(Collections.singletonList(outPut));
    }

    @Override
    public void close() throws Exception {
        if (executorService != null) {
            executorService.shutdown();
        }
        if (resultSet != null) {
            resultSet.close();
        }
        if (preparedStatement != null) {
            preparedStatement.close();
        }
        if (connection != null) {
            connection.close();
        }
    }

}
